package data.platform.event.process;

import data.platform.common.domain.MetricValue;
import data.platform.common.event.CacheLoadedEvent;
import data.platform.common.event.DropMetricValueEvent;
import data.platform.common.event.MetricValueEvent;
import data.platform.common.service.command.DropMetricValuePersistence;
import data.platform.common.service.command.MetricValueCommandService;
import data.platform.config.MetricSaveBufferConfig;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import javax.annotation.PostConstruct;
import java.util.List;

@Component
@Slf4j
@RequiredArgsConstructor
public class MetricValueEventProcess implements ApplicationListener<CacheLoadedEvent> {

    final ApplicationContext applicationContext;

    final DropMetricValuePersistence dropMetricValuePersistence;

    final MetricValueCommandService metricValueCommandService;

    final MeterRegistry dataSetMeterRegistry;

    final MetricSaveBufferConfig metricSaveBufferConfig;

    final Environment environment;

    Counter saveCounter;

    Counter dropCounter;

    private volatile boolean cacheLoaded = false;

    @PostConstruct
    public void init() {

        saveCounter = Counter.builder("dp.metricValue.count.save")
                .register(dataSetMeterRegistry);

        dropCounter = Counter.builder("dp.metricValue.count.drop")
                .register(dataSetMeterRegistry);

        ConfigurableApplicationContext context = (ConfigurableApplicationContext) applicationContext;
        Flux<MetricValueEvent> metricValueEventFlux = Flux.create(fluxSink -> {
            context.addApplicationListener((ApplicationListener<MetricValueEvent>) metricValueEvent -> {
                if (cacheLoaded) {
                    fluxSink.next(metricValueEvent);
                }
            });
        });

        bizProcessPipe(metricValueEventFlux).subscribe();
    }

    private Flux<Integer> bizProcessPipe(Flux<MetricValueEvent> metricValueEventFlux) {
        // -Dreactor.bufferSize.small=20 这个值设置了生产者生产数据后reactor的队列容量上限，达到后执行overflow操作，默认值为256
        // 首先使用onBackpressureBuffer容量，达到上限后，使用reactor.bufferSize.small容量
        // 总的缓存是onBackpressureBuffer+reactor.bufferSize.small
        return metricValueEventFlux
                // maxSize不是队列的长度，是reactor把数据发送消费者前缓存的数据容量
                .onBackpressureBuffer(metricSaveBufferConfig.getMaxSize(),
                        metricEvent -> {
                            log.info("onBackpressureBuffer.....................");
                            // 如果超出容量，数据写入磁盘，以后恢复
                            dropCounter.increment();
                            List<MetricValue> metricValues = metricEvent.getMetricValues();
                            metricValues.forEach(metricValue -> dropMetricValuePersistence.addDropEvent(new DropMetricValueEvent(metricValue)));
                        },
                        BufferOverflowStrategy.DROP_OLDEST)
                .map(MetricValueEvent::getMetricValues)
                //.publishOn(Schedulers.parallel())
                .flatMap(metricValues -> metricValueCommandService.saveAll(metricValues))
                .onErrorContinue((throwable, o) -> log.error("{}, continuing", throwable.getMessage()))
                .doOnNext(count -> saveCounter.increment(count));
    }

    @Override
    public void onApplicationEvent(CacheLoadedEvent event) {
        log.info("cache load finish, start event processing......");
        cacheLoaded = true;
    }
}
